package com.sample.springbatch.job.sample.config;

import com.sample.springbatch.job.config.BaseConfig;
import com.sample.springbatch.job.sample.partitioner.RangePartitioner;
import com.sample.springbatch.job.sample.processor.RepayOrderProcessor;
import com.sample.springbatch.job.sample.reader.PartitionerStepSlaveReader;
import com.sample.springbatch.model.RepayOrderInfo;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.core.partition.PartitionHandler;
import org.springframework.batch.core.partition.support.SimpleStepExecutionSplitter;
import org.springframework.batch.core.partition.support.TaskExecutorPartitionHandler;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.remoting.httpinvoker.HttpInvokerProxyFactoryBean;

import javax.sql.DataSource;
import java.util.Map;

/**
 * 作业分区配置类
 * @author zwg
 * @date 2018/1/12.
 */
@Configuration
public class PartitionerStepConfig extends BaseConfig {


    @Autowired
    RangePartitioner rangePartitioner;

    @Autowired
    DataSource dataSource;


    @Bean(name = "partitionerJob")
    public Job partitionerJob() {
        return jobBuilderFactory.get("partitionerJob")
                .incrementer(new RunIdIncrementer())
                .start(partitionerStepMaster())
                .build();
    }


    @Bean(name = "partitionerStepMaster")
    public Step partitionerStepMaster(){
        //直接配置
        /*return stepBuilderFactory.get("partitionerStepMaster")
                .partitioner("rangePartitioner",rangePartitioner)
                .step(partitionerStepSlave())
                .gridSize(5)
                .taskExecutor(new SimpleAsyncTaskExecutor())
                .build();*/
        //使用handler配置
        return stepBuilderFactory.get("partitionerStepMaster")
                .partitioner("rangePartitioner",rangePartitioner)
                .partitionHandler(partitionHandler())
                .splitter(new SimpleStepExecutionSplitter())
                .build();
    }

    @Bean
    public PartitionHandler partitionHandler() {
        TaskExecutorPartitionHandler retVal = new TaskExecutorPartitionHandler();
        retVal.setTaskExecutor(new SimpleAsyncTaskExecutor());
        retVal.setStep(partitionerStepSlave());
        retVal.setGridSize(5);
        return retVal;
    }

    @Bean(name="remoteStep")
    public HttpInvokerProxyFactoryBean remoteStep(){
        HttpInvokerProxyFactoryBean bean = new HttpInvokerProxyFactoryBean();
        bean.setServiceInterface(Step.class);
        bean.setServiceUrl("");
        return bean;
    }

    @Bean(name = "partitionerStepSlave")
    public Step partitionerStepSlave(){
        return stepBuilderFactory.get("partitionerStepSlave")
                //.tasklet(partitionerStepSlaveTasklet())
                .<RepayOrderInfo, RepayOrderInfo> chunk(100)
                .reader(partitionerStepSlaveReader())
                .processor(partitionerStepSlaveProcessor())
                //.throttleLimit(2)
                //.taskExecutor(new SimpleAsyncTaskExecutor())
                .build();
    }

    @Bean(name = "partitionerStepSlaveReader")
    @StepScope
    public PartitionerStepSlaveReader partitionerStepSlaveReader(){
        return new PartitionerStepSlaveReader();
    }



    @Bean(name = "partitionerStepSlaveProcessor")
    public RepayOrderProcessor partitionerStepSlaveProcessor(){
        return new RepayOrderProcessor();
    }




    @Bean(name = "partitionerStepSlaveTasklet")
    public Tasklet partitionerStepSlaveTasklet() {
        return new Tasklet() {
            @Override
            public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
                Map<String,Object> sec = chunkContext.getStepContext().getStepExecutionContext();
                int fromId = 0;
                int toId = 10;
                String name = "";
                try{
                    fromId = Integer.valueOf(sec.getOrDefault("fromId",0).toString());
                    toId = Integer.valueOf(sec.getOrDefault("toId",0).toString());
                    name = sec.getOrDefault("name","defaultName").toString();

                }catch (Exception e){

                }

                for (int i = fromId; i < toId; i++) {
                    System.out.println("execute step " +  chunkContext.getStepContext().getStepName()
                            + ", execute thread: " + Thread.currentThread().getName()
                            + ", seq: "+ i+" name:"+name+" fromId:"+fromId+" toId:"+toId);
                }

                return RepeatStatus.FINISHED;
            }
        };
    }


}
